package io.github.wanggit.antrpc.server.handler;

import io.github.wanggit.antrpc.client.monitor.report.ZipkinAllMethodTracing;
import io.github.wanggit.antrpc.commons.bean.*;
import io.github.wanggit.antrpc.commons.codec.serialize.ISerializer;
import io.github.wanggit.antrpc.commons.codec.serialize.ISerializerHolder;
import io.github.wanggit.antrpc.commons.codec.serialize.SerializerFactory;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.server.invoker.IRpcRequestBeanInvoker;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ServerReadHandler extends SimpleChannelInboundHandler<RpcProtocol> {

    private final IRpcRequestBeanInvoker rpcRequestBeanInvoker;
    private final ISerializerHolder serializerHolder;

    public ServerReadHandler(
            IRpcRequestBeanInvoker rpcRequestBeanInvoker, ISerializerHolder serializerHolder) {
        this.rpcRequestBeanInvoker = rpcRequestBeanInvoker;
        this.serializerHolder = serializerHolder;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcProtocol msg) throws Exception {
        if (msg.getType() == ConstantValues.HB_TYPE) {
            RpcProtocol rpcProtocol = HeartBeatCreator.create(msg.getCmdId());
            ctx.channel().writeAndFlush(rpcProtocol);
        } else {
            ISerializer serializer =
                    SerializerFactory.getInstance()
                            .getDefaultConfigsSerializerByByteCmd(msg.getSerializer());
            if (null == serializer) {
                serializer = serializerHolder.getSerializer();
            }
            RpcRequestBean requestBean =
                    serializer.deserialize(msg.getData(), RpcRequestBean.class);
            if (log.isDebugEnabled()) {
                log.debug(JsonUtils.toJsonString(requestBean));
            }
            SerialNumberThreadLocal.TraceEntity traceEntity =
                    new SerialNumberThreadLocal.TraceEntity();
            traceEntity.setSerialNumber(requestBean.getSerialNumber());
            traceEntity.setCaller(requestBean.getId());
            traceEntity.setCurrentRequestId(IdGenHelper.getUUID());
            SerialNumberThreadLocal.set(traceEntity);
            RpcResponseBean bean = rpcRequestBeanInvoker.invoke(requestBean);
            if (!requestBean.isOneway()) {
                RpcProtocol protocol = new RpcProtocol();
                protocol.setCmdId(msg.getCmdId());
                protocol.setType(msg.getType());
                protocol.setSerializer(msg.getSerializer());
                protocol.setData(serializer.serialize(bean));
                ctx.channel().writeAndFlush(protocol);
            }
            SerialNumberThreadLocal.clean();
            ZipkinAllMethodTracing.cleanTracing();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (log.isErrorEnabled()) {
            log.error(cause.getMessage(), cause);
        }
        ctx.close();
    }
}
